package p.ithorns.framework.redis.stream;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.util.StringUtils;
import p.ithorns.framework.common.model.ParameterizedTypeImpl;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * RedisStreamListener
 * <p>
 * 消息监听器
 * 消费者实现：以组形式提供独立和共享模式；
 * 1. 独立模式指的是每个消费者都能消费到每条消息；
 * 注意：每个消费者和每条消息是重点；
 * 实现方式：同Stream提供不同的group, groupName不限
 * 2. 共享模式指的是同组内的消费者轮次或者随机消费消息，每次只有一个消费者处理一条消息；
 * 这里的重点是 一个消费者和一条消息
 * 实现方式：同Stream提供相同的group和不同的groupName
 * <p>
 * 子类实现暂时无法使用泛型嵌套
 * 如：List<String>会出现无法转换，但是List就可以
 * 跟使用StringRedisTemplate的StringSerializer反序列化有关
 * <p>
 * 如果是基础类型，可能无法转换具体泛型类 (详见 {@link @org.springframework.data.redis.core.DefaultStreamOperations#isSimpleType}(Line69))
 * <p>
 * <p>
 * TODO 经测试发现非组消费使用有问题
 * 非组消费使用时，同StreamKey多个消息同时发布，【消费端】无法收到全部，有漏读。
 * 初步原因：Spring Redis Stream使用Loop机制读取消息，而不是Block 0来永久阻塞(pollTimeout无法配置<=0);
 * 参考源码 {@link @StreamPollTask#doLoop()}
 * 具体原因还需进一步探查
 * 因此，暂时还是使用消费组比较稳定
 *
 * @author ithorns
 * @version 1.0.0
 * @date 2022/12/10 16:58
 */
public abstract class RedisStreamListener<T> implements StreamListener<String, Record<String, T>> {

    private final static Logger log = LoggerFactory.getLogger(RedisStreamListener.class);

    protected final static String DEFAULT_GROUP = "GROUP";

    protected final static String DEFAULT_GROUP_NAME = "CONSUMER";

    protected RedisStreamTemplate<?> template;

    protected Class<?> targetType;

    protected String group;


    public RedisStreamListener() {
        this.targetType = deSerializable();
        if (isGroupConsumer()) {
            this.group = getStream().toUpperCase();
        }
    }

    public RedisStreamListener(RedisStreamTemplate<?> template) {
        this();
        this.template = template;
    }

    public void setTemplate(@Autowired RedisStreamTemplate<?> template) {
        this.template = template;
    }

    private static Type buildType(Type... types) {
        // ParameterizedTypeImpl inner = new ParameterizedTypeImpl(new Type[]{rawClazz}, null, List.class);
        // ParameterizedTypeImpl outer = new ParameterizedTypeImpl(new Type[]{inner}, null, outClazz);
        ParameterizedType beforeType = null;
        if (types != null && types.length > 0) {
            if (types.length == 1) {
                beforeType = new ParameterizedTypeImpl(new Type[]{types[0]}, null, null);
            } else {
                for (int i = types.length - 1; i > 0; i--) {
                    beforeType = new ParameterizedTypeImpl(new Type[]{beforeType == null ? types[i] : beforeType}, null, types[i - 1]);
                }
            }
        }
        return beforeType;
    }

    /**
     * 消息处理具体实现
     *
     * @param value -值
     */
    public abstract void onMessage(T value);

    public abstract String getStream();

    @Override
    public void onMessage(Record<String, T> message) {
        if (null == message) {
            return;
        }

        String stream = message.getStream();

        if (!StringUtils.hasText(stream)) {
            return;
        }

        RecordId msgId = message.getId();
        try {
            T value = message.getValue();
            // 接收到消息
            log.info("{} => {}:{} ", stream, msgId, value);

            onMessage(value);
            if (!isAutoAck()) {
                // 不是自动确认的，需要手动ack
                template.opsForStream().acknowledge(stream, getGroup(), msgId);
                log.info("ACK: [{}]-[{}]-[{}]", stream, getGroup(), msgId);
            }
        } catch (Exception e) {
            log.error("{}: 消息处理失败", getGroupName(), e);
        }
    }

    /**
     * 处理未ack的消息
     * 需要执行定时任务进行处理。
     * 逐个检查消费者, 看是否有未确认的消息，重新执行
     */
    public void handlePendingMessages() {
        StreamOperations streamOps = template.opsForStream();

        String stream = getStream();
        String group = getGroup();
        PendingMessagesSummary pms = streamOps.pending(stream, group);

        if (null != pms) {
            Map<String, Long> consumers = pms.getPendingMessagesPerConsumer();
            String minId = pms.minMessageId();
            String maxId = pms.maxMessageId();
            for (Map.Entry<String, Long> entry : consumers.entrySet()) {
                // 消费者
                String groupName = entry.getKey();
                // 消费者的pending消息数量
                Long consumerTotalPendingMessages = entry.getValue();
                if (consumerTotalPendingMessages > 0) {
                    // 读取消费者pending队列的前10条记录，从ID=0的记录开始，一直到ID最大值
                    Consumer consumer = Consumer.from(group, groupName);

                    // Range<String> range = Range.closed("0", "+");
                    PendingMessages pendingMessages = streamOps.pending(stream, consumer);

                    // 遍历所有pending消息的详情
                    pendingMessages.forEach(message -> {
                        // 消息的ID
                        RecordId recordId = message.getId();
                        // 消息从消费组中获取，到当前的时间；
                        // 如果一个pending消息的idle time很长，表示这消息，可能是在Ack时发生了异常，
                        // 或者还没来得及Ack，消费者就宕机了，导致一直没有被Ack，当消息发生了转移，它会清零，重新计时
                        Duration elapsedTimeSinceLastDelivery = message.getElapsedTimeSinceLastDelivery();
                        // 消息被获取的次数：可以表示转移的次数，每当一条消息的消费者发生变更的时候，它的值都会+1
                        // 如果一条pending消息的delivery counter值很大，表示它在多个消费者之间进行了多次转移都没法成功消费，就需要人工干预
                        long deliveryCount = message.getTotalDeliveryCount();

                        // TODO 检查elapsedTimeSinceLastDelivery和deliveryCount，如果超出一定限制，可以转移消费者；
                        //  如果再超过一定值，就认为是死信， 需要人工干预
                        log.info("open消息，id={}, elapsedTimeSinceLastDelivery={}, deliveryCount={}",
                                recordId, elapsedTimeSinceLastDelivery, deliveryCount
                        );

                        // 通过streamOperations，直接读取这条pending消息，
                        List<Record<String, T>> result = streamOps.range(stream, Range.rightOpen(minId, maxId));
                        if (null != result) {
                            for (Record<String, T> record : result) {
                                // TODO 这里执行日志输出，模拟的就是【重新】消费逻辑
                                log.info("消费了pending消息：id={}, value={}", record.getId(), record.getValue());
                                this.onMessage(record);
                                // 如果手动消费成功后，往消费组提交消息的ACK
                                Long retVal = streamOps.acknowledge(group, record);
                                log.info("消息Ack，一共Ack了{}条", retVal);
                            }
                        }
                    });
                }
            }
        }
    }

    private void claim(final RecordId recordId, String newGroup) {
        String stream = getStream();
        String group = getGroup();
        template.execute((RedisCallback<Object>) connection ->
                connection.xClaim(stream.getBytes(), group, newGroup, Duration.ofSeconds(5), recordId));
    }

    public String getSimpleName() {
        List<String> names = new ArrayList<>();
        names.add(getStream());
        if (!StringUtils.hasText(getGroup())) {
            names.add(getGroup());
        }
        names.add(getClass().getSimpleName());
        return String.join(":", names);
    }

    private Class<?> deSerializable() {
        Type type = getClass().getGenericSuperclass();
        if (type instanceof ParameterizedType) {
            ParameterizedType parameterizedType = (ParameterizedType) type;
            Type actType = parameterizedType.getActualTypeArguments()[0];
            return checkType(actType, 0);
        } else {
            String className = type == null ? "null" : type.getClass().getName();
            throw new IllegalArgumentException("Expected a Class, ParameterizedType"
                    + ", but <" + type + "> is of type " + className);
        }
    }

    private static Class<?> checkType(Type type, int index) {
        if (type instanceof Class<?>) {
            return (Class<?>) type;
        } else if (type instanceof ParameterizedType) {
            ParameterizedType pt = (ParameterizedType) type;
            // Type t = pt.getActualTypeArguments()[index];
            Type rawType = pt.getRawType();
            return checkType(rawType, index);
        } else {
            String className = type == null ? "null" : type.getClass().getName();
            throw new IllegalArgumentException("Expected a Class, ParameterizedType"
                    + ", but <" + type + "> is of type " + className);
        }
    }

    /**
     * 获取目标消息类型
     *
     * @return 具体类型
     */
    public Class<?> getTargetType() {
        //deSerializable();
        return this.targetType;
    }

    /**
     * 是否消费组
     * 默认true
     *
     * @return boolean
     */
    public boolean isGroupConsumer() {
        return true;
    }


    /**
     * 消费组名称
     * 使用同一个组名，就是共享消费
     * 如果需要扇形发布，每个消费组都必须使用自己独立的组名
     *
     * @return String
     */
    public String getGroup() {
        return group;
    }

    public String getGroupName() {
        return getGroup();
    }

    /**
     * 是否自动确认
     *
     * @return boolean
     */
    public boolean isAutoAck() {
        return true;
    }
}